package com.shujia.opt

import org.apache.spark.SparkContext
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.SparkSession

import java.text.SimpleDateFormat
import java.util.Date

object Demo3MapPartitions {
  def main(args: Array[String]): Unit = {
    //使用mapPartitions替代普通map Transformation算子
    //使用reduceByKey/aggregateByKey替代groupByKey
    val sparkSession: SparkSession = SparkSession.builder()
      .config("spark.sql.shuffle.partitions", "1")
      .master("local[2]")
      .appName("缓存优化")
      .getOrCreate()

    val sparkContext: SparkContext = sparkSession.sparkContext

    val lineRDD: RDD[String] = sparkContext.textFile("spark/data/ant_user_low_carbon.txt")
    println(s"lineRDD的分区数：${lineRDD.getNumPartitions}")

    /**
     * map算子主要作用是，遍历RDD中的每一条数据，进行处理返回新的一条数据
     * 如果在处理过程中，需要创建工具对象的话，那么使用map不太好，原因是因为每一条数据都需要new一下
     * 可能会造成内存溢出
     */
//    val resRDD: RDD[(String, String, String)] = lineRDD.map((line: String) => {
//      println("===================创建一次对象=============================")
//      val info: Array[String] = line.split("\t")
//      val t1: String = info(1)
//      val sdf = new SimpleDateFormat("yyyy/MM/dd")
//      val date: Date = sdf.parse(t1)
//      val sdf2 = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss")
//      val t2: String = sdf2.format(date)
//      (info(0), t2, info(2))
//    })
//    resRDD.foreach(println)

    /**
     * 实际上针对上面的案例，我们可以针对rdd中的每一个分区创建一个工具对象，在每条数据上使用
     * mapPartitions,将每一个分区中的数据封装成了一个迭代器
     */
    val resRDD: RDD[(String, String, String)] = lineRDD.mapPartitions((itr: Iterator[String]) => {
      println("===================创建一次对象=============================")
      val sdf = new SimpleDateFormat("yyyy/MM/dd")
      val sdf2 = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss")
      itr.map((line: String) => {
        val info: Array[String] = line.split("\t")
        val t1: String = info(1)
        val date: Date = sdf.parse(t1)
        val t2: String = sdf2.format(date)
        (info(0), t2, info(2))
      })
    })

    resRDD.foreach(println)
  }
}
